package com.shujia.road

import java.sql.{Connection, PreparedStatement}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.shujia.flink.FlinkTool
import com.shujia.road.RealTimeRoadAvgSpeed.env
import com.shujia.util.{CarUtil, DateUtil, JdbcUtil}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

/**
  *
  * 2.1 实时统计每个道路当天总流量
  *
  */
object RealTimeRoadDayFlow extends FlinkTool {
  override def run(args: Array[String]): Unit = {
    //读取数据
    val carsDS: DataStream[String] = CarUtil.loadKafkaData(env)

    //1|解析数据
    val kvDS: DataStream[(Long, String, Int)] = carsDS.map(car => {
      val carJSON: JSONObject = JSON.parseObject(car)
      val road_id: Long = carJSON.getLong("road_id")
      val time: Long = carJSON.getLong("time")

      //将时间转换成日期
      val day: String = DateUtil.tsToDate(time)


      (road_id, day, 1)
    })

    //安装道路和日期分组
    val keyByDS: KeyedStream[(Long, String, Int), (Long, String)] = kvDS.keyBy(kv => (kv._1, kv._2))

    //统计车流量
    val flowDS: DataStream[(Long, String, Int)] = keyByDS.sum(2)

    /**
      * 将结果保存到mysql
      *
      */
    flowDS.addSink(new RichSinkFunction[(Long, String, Int)] {

      var con: Connection = _

      override def open(parameters: Configuration): Unit = {

        //获取链接
        con = JdbcUtil.getConnection
      }

      override def close(): Unit = {
        con.close()
      }

      override def invoke(value: (Long, String, Int), context: SinkFunction.Context[_]): Unit = {
        val stat: PreparedStatement = con.prepareStatement("replace into real_time_road_day_flow(road_id,day,flow) values(?,?,?)")

        stat.setLong(1, value._1)
        stat.setString(2, value._2)
        stat.setLong(3, value._3)

        stat.execute()

      }
    })


    env.execute("RealTimeRoadDayFlow$")


  }
}
